package com.sh.data.engine.domain.datadev.offline.runner;

import cn.hutool.extra.spring.SpringUtil;
import com.google.common.collect.Lists;
import com.sap.db.jdbc.HanaClob;
import com.sh.data.engine.common.constants.Constants;
import com.sh.data.engine.domain.base.service.LogService;
import com.sh.data.engine.domain.dataasset.service.TableProductService;
import com.sh.data.engine.domain.datadev.offline.config.OfflineDevConfig;
import com.sh.data.engine.domain.datadev.offline.model.enums.OfflineDevJobStatusEnum;
import com.sh.data.engine.domain.datadev.offline.service.DatadevJobService;
import com.sh.data.engine.domain.util.ClobUtil;
import com.sh.data.engine.infrastructure.config.DataXConfig;
import com.sh.data.engine.infrastructure.config.FileStorageConfiguration;
import com.sh.data.engine.infrastructure.config.HDFSConfig;
import com.sh.data.engine.infrastructure.config.MapReduceConfiguration;
import com.sh.data.engine.repository.dao.datadev.entity.DatadevJobEntity;
import com.sh.data.engine.repository.dao.datadev.entity.DatadevMenuNodeEntity;
import gudusoft.gsqlparser.EDbVendor;
import gudusoft.gsqlparser.TCustomSqlStatement;
import gudusoft.gsqlparser.TGSqlParser;
import gudusoft.gsqlparser.TStatementList;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang.StringEscapeUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.commons.lang3.time.DateFormatUtils;

import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.stream.Collectors;

/**
 * @Author: zigui.zdf @Date: 2022/3/1 10:55:56 @Description:
 */
@Data
@Slf4j
public abstract class OfflineDevBaseJobRunner implements Runnable {

    /**
     * 当前任务记录
     */
    protected DatadevJobEntity jobEntity;

    protected DatadevMenuNodeEntity menuNodeEntity;

    /**
     * 父节点当前节点的连线类型 0-执行成功后执行 1-执行失败后执行 2-执行失败成功都执行
     */
    protected Map<Long, Integer> parentNodeIdAndLineStatusMap;

    /**
     * 当前节点的所有儿子节点
     */
    protected List<Long> childNodeIdList;

    /**
     * 执行器
     */
    protected ExecutorService executorService;

    /**
     * 工作流所有节点的map集合，key是nodeId
     */
    protected Map<Long, OfflineDevBaseJobRunner> allJobNodeMap;

    protected LogService logService = SpringUtil.getBean(LogService.class);

    protected DatadevJobService datadevJobService = SpringUtil.getBean(DatadevJobService.class);

    protected TableProductService tableProductService = SpringUtil.getBean(TableProductService.class);

    protected FileStorageConfiguration fileStorageConfiguration = SpringUtil.getBean(FileStorageConfiguration.class);

    protected OfflineDevConfig offlineDevConfig = SpringUtil.getBean(OfflineDevConfig.class);

    protected DataXConfig dataXConfig = SpringUtil.getBean(DataXConfig.class);

    protected MapReduceConfiguration mapReduceConfiguration = SpringUtil.getBean(MapReduceConfiguration.class);

    protected HDFSConfig hdfsConfig = SpringUtil.getBean(HDFSConfig.class);

    /**
     * 任务是否停止
     */
    protected boolean isStop = false;

    /**
     * 任务是否结束，不再执行
     */
    protected boolean isFinished = false;

    public OfflineDevBaseJobRunner(
        DatadevJobEntity jobEntity,
        DatadevMenuNodeEntity menuNodeEntity,
        Map<Long, Integer> parentNodeIdAndLineStatusMap,
        List<Long> childNodeIdList,
        Map<Long, OfflineDevBaseJobRunner> allJobNodeMap,
        ExecutorService executorService) {
        this.jobEntity = jobEntity;
        this.menuNodeEntity = menuNodeEntity;
        this.parentNodeIdAndLineStatusMap = parentNodeIdAndLineStatusMap;
        this.childNodeIdList = childNodeIdList;
        this.executorService = executorService;
        this.allJobNodeMap = allJobNodeMap;
    }

    @Override
    public void run() {
        if (isStop) {
            isFinished = true;
            return;
        }

        log.info("节点:[{}]开始执行...", menuNodeEntity.getName());
        String format = String.format("节点:[%s]开始执行...", menuNodeEntity.getName());
        writeLog(Lists.newArrayList(format));
        jobEntity.setStartTime(new Date());
        jobEntity.setStatus(OfflineDevJobStatusEnum.RUNNING.getCode());
        jobEntity.setPlanTime(new Date());
        datadevJobService.updateById(jobEntity);

        long s = System.currentTimeMillis();

        int status;
        try {
            status = runJob();
        } catch (Exception e) {
            if (!isStop) {
                List<String> logs = new ArrayList<>();
                logs.add(String.format("任务:[%s]执行异常", menuNodeEntity.getName()));
                final String stackTrace = ExceptionUtils.getStackTrace(e);
                logs.add(stackTrace);
                writeLog(logs);
                status = OfflineDevJobStatusEnum.FAILURE.getCode();
            } else {
                status = OfflineDevJobStatusEnum.CANCEL.getCode();
            }
        }

        jobEntity.setEndTime(new Date());
        jobEntity.setStatus(status);
        datadevJobService.updateById(jobEntity);

        isFinished = true;

        String finishLog =
            String.format(
                "节点:[%s]执行结束,任务状态:[%s],耗时:[%s]ms \r\n",
                menuNodeEntity.getName(), jobEntity.getStatus(), System.currentTimeMillis() - s);

        writeLog(Lists.newArrayList(finishLog));
        log.info(
            "节点:[{}]执行结束,任务状态:[{}],耗时:[{}]ms",
            menuNodeEntity.getName(),
            jobEntity.getStatus(),
            System.currentTimeMillis() - s);

        runChildJob();

        updateTaskFlowJobStatus();
    }

    /**
     * 停止任务
     */
    public void stopJob() {
        if (!isFinished) {
            String logStop = String.format("节点[%s]即将停止运行......", menuNodeEntity.getName());
            String format = formatLogWithTime(logStop);
            writeLog(Lists.newArrayList(format));
        }

        this.isStop = true;

        new Thread(
            () -> {
                this.cancel();

                String stopSuccess = String.format("节点[%s]停止运行成功", menuNodeEntity.getName());
                String format = formatLogWithTime(stopSuccess);
                writeLog(Lists.newArrayList(format));
            })
            .start();
    }

    public abstract void cancel();

    /**
     * 执行任务
     */
    public abstract int runJob() throws Exception;

    public void writeLog(List<String> lines) {
        if (CollectionUtils.isEmpty(lines)) {
            return;
        }

        final String dir = fileStorageConfiguration.getDir();

        if (null != jobEntity.getTaskFlowJobId()) {
            String taskFlowJobLogPath =
                String.format(
                    "%s/log/%s/%s.log",
                    dir, Constants.LogFileExp.OFFLINE_DEV_JOB_LOG_EXP, jobEntity.getTaskFlowJobId());
            logService.write(taskFlowJobLogPath, lines);
        }

        String subTaskJobLogPath =
            String.format(
                "%s/log/%s/%s.log",
                dir, Constants.LogFileExp.OFFLINE_DEV_JOB_LOG_EXP, jobEntity.getId());

        logService.write(subTaskJobLogPath, lines);
    }

    public void writeRs(List<List<String>> lines, int rsIndex) {
        if (CollectionUtils.isEmpty(lines)) {
            return;
        }

        final String dir = fileStorageConfiguration.getDir();

        String rsPath;
        if (Objects.nonNull(jobEntity.getTaskFlowJobId())) {
            rsPath =
                String.format(
                    "%s/log/%s/%s/%s%s",
                    dir,
                    Constants.LogFileExp.OFFLINE_DEV_JOB_RS_EXP,
                    jobEntity.getTaskFlowJobId(),
                    rsIndex,
                    Constants.LogFileExp.RS_TMP_FILE_SUFFIX);
        } else {
            rsPath =
                String.format(
                    "%s/log/%s/%s/%s%s",
                    dir,
                    Constants.LogFileExp.OFFLINE_DEV_JOB_RS_EXP,
                    jobEntity.getId(),
                    rsIndex,
                    Constants.LogFileExp.RS_TMP_FILE_SUFFIX);
        }
        List<String> list = new LinkedList<>();

        for (List<String> line : lines) {
            list.add(
                line.stream()
                    .map(_e -> StringEscapeUtils.escapeCsv(StringEscapeUtils.escapeJava(_e)))
                    .collect(Collectors.joining(Constants.LogFileExp.COL_SEPARATOR)));
        }

        logService.write(rsPath, list);
    }

    public void writeSQL(String sql, int rsIndex) {
        if (StringUtils.isBlank(sql)) {
            return;
        }
        final String dir = fileStorageConfiguration.getDir();
        String rsPath =
            String.format(
                "%s/log/%s/%s/%s%s",
                dir,
                Constants.LogFileExp.OFFLINE_DEV_JOB_RS_EXP,
                jobEntity.getId(),
                rsIndex,
                Constants.LogFileExp.SQL_FILE_SUFFIX);
        logService.write(rsPath, Lists.newArrayList(sql));
    }

    public void toResult(ResultSet rs, int rsIndex, String sql) throws SQLException {
        ResultSetMetaData meta = rs.getMetaData();

        int columnCount = meta.getColumnCount();

        List<List<String>> headers = new LinkedList<>();
        // 第一行是列名
        List<String> header = new LinkedList<>();
        for (int column = 1; column <= columnCount; column++) {
            String columnName = meta.getColumnLabel(column);
            columnName = columnName.substring(columnName.lastIndexOf(".") + 1);
            header.add(columnName);
        }
        headers.add(header);

        writeRs(headers, rsIndex);

        final int resultLimit = offlineDevConfig.getResultLimit();

        List<List<String>> list = new LinkedList<>();
        // 其他行是每列的数据
        int index = 0;
        while (rs.next()) {
            if (resultLimit > 0 && index >= resultLimit) {
                break;
            }
            List<String> line = new LinkedList<>();
            for (int column = 1; column <= columnCount; column++) {
                line.add(objToString(rs.getObject(column)));
            }
            list.add(line);

            if (list.size() >= 100) {
                writeRs(list, rsIndex);
                list.clear();
            }
            index++;
        }
        if (list.size() > 0) {
            writeRs(list, rsIndex);
            list.clear();
        }
        writeSQL(sql, rsIndex);
    }

    private String objToString(Object obj) throws SQLException {
        if (null == obj) {
            return "\\N";
        }
        if (obj instanceof HanaClob) {
            return ClobUtil.hanaClobToString2((HanaClob) obj);
        }
        return Objects.toString(obj, "\\N");
    }

    public String formatLogWithTime(String log) {
        String time = DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss.SSS");
        return String.format("[%s] %s", time, log);
    }

    /**
     * 执行儿子节点
     */
    private void runChildJob() {
        if (CollectionUtils.isEmpty(childNodeIdList)) {
            log.info("当前节点:[{}]无子节点，结束", menuNodeEntity.getName());
            return;
        }

        for (Long childNodeId : childNodeIdList) {
            final OfflineDevBaseJobRunner childNode = allJobNodeMap.get(childNodeId);

            if (isStop) {
                childNode.setFinished(true);
                setChildNodesFinished(childNodeId);
                return;
            }

            if (childNode.isFinished) {
                continue;
            }

            boolean childJobNodeCanRun = true;

            final Map<Long, Integer> parentNodeIdAndLineStatusMap =
                childNode.getParentNodeIdAndLineStatusMap();

            // 判断父节点有没有执行结束
            if (null != parentNodeIdAndLineStatusMap && !parentNodeIdAndLineStatusMap.isEmpty()) {
                for (Map.Entry<Long, Integer> entry : parentNodeIdAndLineStatusMap.entrySet()) {
                    final Long parentNodeId = entry.getKey();
                    final Integer lineStatus = entry.getValue();
                    final OfflineDevBaseJobRunner parentNode = allJobNodeMap.get(parentNodeId);
                    final Integer status = parentNode.getJobEntity().getStatus();

                    // 如果父级任务还在运行，则直接跳过
                    if (status == OfflineDevJobStatusEnum.RUNNING.getCode()
                        || status == OfflineDevJobStatusEnum.NO_RUN.getCode()) {
                        childJobNodeCanRun = false;
                        log.info(
                            "当前节点:[{}]的子节点:[{}],因其中一个父节点:[{}]还在执行，暂不执行该子任务",
                            menuNodeEntity.getName(),
                            childNode.getMenuNodeEntity().getName(),
                            parentNode.getMenuNodeEntity().getName());
                        break;
                    }

                    // 如果父级任务状态和连线状态不匹配，则子任务将无法再运行，设置子任务状态及该子任务的所有下游任务的isFinished = true
                    final boolean match = isMatch(status, lineStatus);
                    if (!match) {
                        childJobNodeCanRun = false;
                        childNode.setFinished(true);
                        setChildNodesFinished(childNodeId);
                        log.info(
                            "当前节点:[{}]的子节点:[{}],因其中一个父节点:[{}]的任务状态和连线状态不匹配，不再执行该子任务",
                            menuNodeEntity.getName(),
                            childNode.getMenuNodeEntity().getName(),
                            parentNode.getMenuNodeEntity().getName());
                        break;
                    }
                }
            }

            if (childJobNodeCanRun && !isStop) {
                executorService.submit(childNode);
            }
        }
    }

    private void setChildNodesFinished(Long parentNodeId) {
        final OfflineDevBaseJobRunner jobRunner = allJobNodeMap.get(parentNodeId);

        final List<Long> childNodeIdList = jobRunner.getChildNodeIdList();

        if (CollectionUtils.isEmpty(childNodeIdList)) {
            return;
        }

        childNodeIdList.forEach(
            childNodeId -> {
                final OfflineDevBaseJobRunner childNode = allJobNodeMap.get(childNodeId);
                childNode.setFinished(true);
                setChildNodesFinished(childNodeId);
            });
    }

    /**
     * 更新工作流的状态
     */
    private void updateTaskFlowJobStatus() {
        // 判断所有子节点的isFinished是否都为true，是的话更新工作流的状态
        boolean isFinished = true;

        Set<Integer> subTaskStatusSet = new HashSet<>();

        for (Map.Entry<Long, OfflineDevBaseJobRunner> entry : allJobNodeMap.entrySet()) {
            final Integer status = entry.getValue().getJobEntity().getStatus();
            final boolean finished = entry.getValue().isFinished;
            if (!finished) {
                isFinished = false;
                break;
            }
            subTaskStatusSet.add(status);
        }

        if (!isFinished) {
            return;
        }

        int taskFlowJobStatus;

        if (subTaskStatusSet.contains(OfflineDevJobStatusEnum.CANCEL.getCode())) {
            taskFlowJobStatus = OfflineDevJobStatusEnum.CANCEL.getCode();
        } else if (subTaskStatusSet.contains(OfflineDevJobStatusEnum.SUCCESS.getCode())
            && subTaskStatusSet.contains(OfflineDevJobStatusEnum.FAILURE.getCode())) {
            taskFlowJobStatus = OfflineDevJobStatusEnum.PART_FAILED.getCode();
        } else if (subTaskStatusSet.contains(OfflineDevJobStatusEnum.SUCCESS.getCode())
            && !subTaskStatusSet.contains(OfflineDevJobStatusEnum.FAILURE.getCode())) {
            taskFlowJobStatus = OfflineDevJobStatusEnum.SUCCESS.getCode();
        } else {
            taskFlowJobStatus = OfflineDevJobStatusEnum.FAILURE.getCode();
        }

        datadevJobService.updateStatusFinished(
            jobEntity.getTaskFlowJobId(), taskFlowJobStatus, false, true);
    }

    /**
     * 0-执行成功后执行 1-执行失败后执行 2-执行失败成功都执行
     *
     * @param jobStatus
     * @param lineStatus
     * @return
     */
    private boolean isMatch(Integer jobStatus, Integer lineStatus) {
        final OfflineDevJobStatusEnum jobStatusEnum = OfflineDevJobStatusEnum.of(jobStatus);

        if (OfflineDevJobStatusEnum.SUCCESS.equals(jobStatusEnum)) {
            // 0-执行成功后执行
            if (lineStatus == 0) {
                return true;
            }

            // 2-执行失败成功都执行
            if (lineStatus == 2) {
                return true;
            }
        }

        if (OfflineDevJobStatusEnum.FAILURE.equals(jobStatusEnum)) {
            // 1-执行失败后执行
            if (lineStatus == 1) {
                return true;
            }

            // 2-执行失败成功都执行
            if (lineStatus == 2) {
                return true;
            }
        }

        return false;
    }

    public static void main(String[] args) {
        String sql =
            "select * from HUFU.\"ods_tererrerr\";\n"
                + "select * from HUFU.\"ods_tererrerr\";\n"
                + "select * from HUFU.\"ods_tererrerr\";\n"
                + "select * from HUFU.\"ods_tererrerr\";\n"
                + "select * from HUFU.\"ods_tererrerr\";\n"
                + "select * from HUFU.\"ods_tererrerr\";\n"
                + "select * from HUFU.\"ods_tererrerr\";\n"
                + "select * from HUFU.\"ods_tererrerr\";\n"
                + "select * from HUFU.\"ods_tererrerr\";\n"
                + "select * from HUFU.\"ods_tererrerr\";\n"
                + "select * from HUFU.\"ods_tererrerr\";\n"
                + "select * from HUFU.\"ods_tererrerr\";"; // 删表;

        String sql2 =
            "--********************************************************************--\\n--author: admin\\n--create time: Thu Apr 21 2022 13:45:35 GMT+0800 (中国标准时间)\\n--********************************************************************--\\nCREATE COLUMN TABLE \"CFDATA\".\"DWS_QM_VEHINFO_FEED_TOTAL_R\" (\"FAREAID\" NVARCHAR(10) COMMENT '市场',\n"
                + "\"FFACTORY\" NVARCHAR(50) COMMENT '工厂',\n"
                + "\"FCATEGORY\" NVARCHAR(50) COMMENT '车辆类型（两轮/四轮）',\n"
                + "\"FSERIES\" NVARCHAR(50) COMMENT '车系列',\n"
                + "\"FCARTYPE\" NVARCHAR(50) COMMENT '车型',\n"
                + "\"FSUPPLIER\" NVARCHAR(255) COMMENT '供应商',\n"
                + "\"FSUPPLIERNO\" VARCHAR(50) COMMENT '供应商编号',\n"
                + "\"FMATERIAL\" NVARCHAR(255) COMMENT '投料编号',\n"
                + "\"FMATERIALNAME\" VARCHAR(255) COMMENT '投料名称',\n"
                + "\"FBASEDATE\" VARCHAR(10) COMMENT '基准日期',\n"
                + "\"FDTMONTH\" VARCHAR(10) COMMENT '倒推月份',\n"
                + "\"FFEED_QTY\" DECIMAL(18,\n"
                + "4) CS_FIXED COMMENT '投料数量',\n"
                + "\"INSERT_TIME\" LONGDATE CS_LONGDATE COMMENT '固化时间') UNLOAD PRIORITY 5 AUTO MERGE COMMENT 'QIS:车型/物料/供应商投料汇总(倒推月份)-ppm';select * from HUFU.\\\"ods_tererrerr\\\";";

        String sql3 =
            "--********************************************************************--\n"
                + "--author: admin\n"
                + "--create time: Thu Apr 21 2022 15:44:16 GMT+0800 (中国标准时间)\n"
                + "--********************************************************************--\n"
                + "\n"
                + "CREATE COLUMN TABLE \"CFDATA\".\"DWS_QM_VEHINFO_FEED_TOTAL_R\" (\"FAREAID\" NVARCHAR(10) COMMENT '市场',\n"
                + "\"FFACTORY\" NVARCHAR(50) COMMENT '工厂',\n"
                + "\"FCATEGORY\" NVARCHAR(50) COMMENT '车辆类型（两轮/四轮）',\n"
                + "\"FSERIES\" NVARCHAR(50) COMMENT '车系列',\n"
                + "\"FCARTYPE\" NVARCHAR(50) COMMENT '车型',\n"
                + "\"FSUPPLIER\" NVARCHAR(255) COMMENT '供应商',\n"
                + "\"FSUPPLIERNO\" VARCHAR(50) COMMENT '供应商编号',\n"
                + "\"FMATERIAL\" NVARCHAR(255) COMMENT '投料编号',\n"
                + "\"FMATERIALNAME\" VARCHAR(255) COMMENT '投料名称',\n"
                + "\"FBASEDATE\" VARCHAR(10) COMMENT '基准日期',\n"
                + "\"FDTMONTH\" VARCHAR(10) COMMENT '倒推月份',\n"
                + "\"FFEED_QTY\" DECIMAL(18,\n"
                + "4) CS_FIXED COMMENT '投料数量',\n"
                + "\"INSERT_TIME\" LONGDATE CS_LONGDATE COMMENT '固化时间') UNLOAD PRIORITY 5 AUTO MERGE COMMENT 'QIS:车型/;物料/供应商投料汇总(倒推月份)-ppm'\n"
                + "\n"
                + "SELECT * from \"CFDATA\".\"DWS_QM_VEHINFO_FEED_TOTAL_R\"\n";
        TGSqlParser sqlParser = new TGSqlParser(EDbVendor.dbvhana);
        sqlParser.sqltext = sql3;
        int parse = sqlParser.parse();
        TStatementList stList = sqlParser.sqlstatements;
        for (int i = 0; i < stList.size(); i++) {
            TCustomSqlStatement customSt = stList.get(i);
            String tsql = customSt.toString();
            System.out.println("第" + (i + 1) + "条sql:" + tsql);
        }
        System.out.println("共有条" + stList.size() + "sql语句!");
        System.out.println(parse);

        System.out.println(sqlParser.getErrormessage());
    }
}
